package com.fengye.provider;

import com.alibaba.fastjson.JSON;
import com.fengye.model.ProviderService;
import com.fengye.model.RemoteRequest;
import com.fengye.model.RemoteResponse;
import com.fengye.register.RegisterCenter;
import com.fengye.register.RegisterCenter4Provider;
import com.google.common.base.Predicate;
import com.google.common.collect.Collections2;
import com.google.common.collect.Maps;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.Method;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

/**
 * 服务调用实现类
 *
 * @Author fengye
 * @Date 2023/2/13 14:21
 **/
public class NettyServerInvokeHandler extends SimpleChannelInboundHandler<RemoteRequest> {

    private final static Logger logger = LoggerFactory.getLogger(NettyServer.class);

    /** 服务端限流工具 */
    private final static Map<String, Semaphore> serviceKeySemaphoreMap = Maps.newConcurrentMap();

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.flush();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        logger.error(cause.getMessage());
        cause.printStackTrace();
        // 发生异常，关闭链路
        ctx.close();
    }

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, RemoteRequest remoteRequest) throws Exception {
        if(remoteRequest == null || remoteRequest.getProviderService() == null) {
            logger.error("channel read error. remoteRequest = " + remoteRequest);
            return;
        }

        if(channelHandlerContext.channel().isWritable()) {
            // 从服务调用方里获取服务提供者信息
            ProviderService providerService = remoteRequest.getProviderService();
            long consumeTimeout = remoteRequest.getInvokeTimeout();
            final String methodName = remoteRequest.getInvokedMethodName();
            // 根据方法名称定位具体某一个服务提供者
            String serviceKey = providerService.getServiceItf().getName();
            // 获取限流数,根据工作数
            int workerThread = providerService.getWorkerThread();
            Semaphore semaphore = getSemaphore(serviceKey, workerThread);
            // 双重检查
            if(semaphore == null) {
                synchronized (serviceKeySemaphoreMap) {
                    semaphore = serviceKeySemaphoreMap.get(serviceKey);
                    if (semaphore == null) {
                        semaphore = new Semaphore(workerThread);
                        serviceKeySemaphoreMap.put(serviceKey, semaphore);
                    }
                }
            }

            //获取注册服务中心
            RegisterCenter4Provider registerCenter4Provider = RegisterCenter.singleton();
            List<ProviderService> localProviderCaches = registerCenter4Provider.getProviderServiceMap().get(serviceKey);

            Object result = null;
            boolean acquire = false;
            boolean success = true;
            try {
                ProviderService localProviderCache = Collections2.filter(localProviderCaches, new Predicate<ProviderService>() {
                    @Override
                    public boolean apply(ProviderService input) {
                        return StringUtils.equals(input.getServiceMethod().getName(), methodName);
                    }
                }).iterator().next();
                // 利用反射发起服务
                Object serviceObject = localProviderCache.getServiceObject();
                Method method = localProviderCache.getServiceMethod();

                // 利用semaphore限流
                acquire = semaphore.tryAcquire(consumeTimeout, TimeUnit.MICROSECONDS);
                if (acquire) {
                    // 调用方法
                    result = method.invoke(serviceObject, remoteRequest.getArgs());
                    logger.info("result:" + JSON.toJSONString(result));
                }
            } catch (Exception e) {
                logger.error("localProviderCaches:" + JSON.toJSONString(localProviderCaches)
                        + ", methodName: " + methodName + " invoke error." + e.getMessage());
                result = e;
                success = false;
            } finally {
                // 调用结束，释放资源
                if(acquire) {
                    semaphore.release();
                }
            }

            // 根据服务调用结果组装结果返回对象
            RemoteResponse remoteResponse = new RemoteResponse();
            remoteResponse.setInvokeTimeout(consumeTimeout);
            remoteResponse.setUniqueKey(remoteRequest.getUniqueKey());
            remoteResponse.setResult(result);
            remoteResponse.setSuccess(success);
            // 将服务调用返回对象回写到消费端
            channelHandlerContext.writeAndFlush(remoteResponse);
        } else {
            logger.error("remoteRequest:" + JSON.toJSONString(remoteRequest) + " is error.");
        }
    }

    /**
     * 获取semaphore，用于限流
     * @param serviceKey
     * @param workerThread
     * @return
     */
    private Semaphore getSemaphore(String serviceKey, int workerThread) {
        Semaphore semaphore = serviceKeySemaphoreMap.get(serviceKey);
        // 双重检查，保证线程安全
        if (semaphore == null) {
            synchronized (serviceKeySemaphoreMap) {
                semaphore = serviceKeySemaphoreMap.get(serviceKey);
                if(semaphore == null) {
                    semaphore = new Semaphore(workerThread);
                    serviceKeySemaphoreMap.put(serviceKey,semaphore);
                }
            }
        }
        return semaphore;
    }
}
